#ifndef __M_CONSUMER_H__
#define __M_CONSUMER_H__
#include "../mqCommon/logger.hpp"
#include "../mqCommon/helper.hpp"
#include "../mqCommon/mq_msg.pb.h"
#include <iostream>
#include <memory>
#include <mutex>
#include <functional>
#include <unordered_map>

/*
    1.消费者是以队列为单位管理起来的
    2.每个用户订阅一个队列时就会在信道上创建一个消费者，这个消费者连接了信道和队列，
      队列发布消息就是发布给消费者，再由消费者传回信道，所以管理消费者是以队列为单位，
      当然信道删除时也会清空上面的消费者
*/
namespace yqx{
    using ConsumerCallback = std::function<void(const std::string, const BasicProperties*, const std::string)>;
    struct Consumer{
        using ptr = std::shared_ptr<Consumer>;
        std::string _tag;//消费者标识
        std::string _qname;//消费者订阅的队列名称
        bool _auto_ack;//自动确认标识
        ConsumerCallback _callback;

        Consumer(){}
        Consumer(const std::string& tag, const std::string& qname, bool auto_ack, const ConsumerCallback& cb):
        _tag(tag), _qname(qname), _auto_ack(auto_ack), _callback(std::move(cb))
        {}
    };
    //以队列为单元的消费者管理结构
    class QueueConsumer{
        public:
            using ptr = std::shared_ptr<QueueConsumer>;
            QueueConsumer(const std::string& qname):_qname(qname), _rr_seq(0){}
            //队列新增消费者
            Consumer::ptr create(const std::string& tag, const std::string& qname, bool auto_ack, const ConsumerCallback& cb){
                std::unique_lock<std::mutex> lock(_mtx);
                //判断消费者是否重复
                for(auto cs : _consumers){
                    if(cs->_tag == tag)
                        return nullptr;
                }
                auto cs = std::make_shared<Consumer>(tag, qname, auto_ack, cb);
                _consumers.push_back(cs);
                return cs;
            }
            //移除某个消费者
            void remove(const std::string& tag){
                std::unique_lock<std::mutex> lock(_mtx);
                for(auto it = _consumers.begin(); it != _consumers.end(); it++){
                    if((*it)->_tag == tag){
                        _consumers.erase(it);
                        return;
                    }
                }
            }
            //队列获取消费者：RR轮转获取
            Consumer::ptr choose(){
                std::unique_lock<std::mutex> lock(_mtx);
                if(_consumers.size() == 0)
                    return nullptr;
                _rr_seq %= _consumers.size();
                int idx = _rr_seq;
                _rr_seq++;
                return _consumers[idx];
            }
            //判断队列是否为空
            bool empty(){
                std::unique_lock<std::mutex> lock(_mtx);
                return _consumers.size() == 0;
            }
            //判断某个消费者是否存在
            bool exists(const std::string& tag){
                std::unique_lock<std::mutex> lock(_mtx);
                for(auto it = _consumers.begin(); it != _consumers.end(); it++){
                    if((*it)->_tag == tag){
                        return true;
                    }
                }
                return false;
            }
            //清空所有消费者
            void clear(){
                std::unique_lock<std::mutex> lock(_mtx);
                _consumers.clear();
            }
        private:
            std::string _qname;
            std::mutex _mtx;
            uint64_t _rr_seq;
            std::vector<Consumer::ptr> _consumers;
    };
    //对外的消费者队列管理句柄
    class ConsumerManager{
        public:
            using ptr = std::shared_ptr<ConsumerManager>;
            ConsumerManager(){}
            //新增一个队列消费者管理句柄
            void initQueueConsumer(const std::string& qname){
                std::unique_lock<std::mutex> lock(_mtx);
                auto it = _qconsumers.find(qname);
                if(it != _qconsumers.end())
                    return;
                QueueConsumer::ptr qcp = std::make_shared<QueueConsumer>(qname);
                _qconsumers.insert(std::make_pair(qname, qcp));
            }
            //删除一个队列消费者管理句柄
            void destroyQueueConsumer(const std::string& qname){
                std::unique_lock<std::mutex> lock(_mtx);
                _qconsumers.erase(qname);
            }
            //创建一个消费者
            Consumer::ptr create(const std::string& tag, const std::string& qname, bool auto_ack, const ConsumerCallback& cb){
                QueueConsumer::ptr qcp;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    auto it = _qconsumers.find(qname);
                    if(it == _qconsumers.end()){
                        LOG("没有找到%s队列的消费者管理句柄!", qname.c_str());
                        return nullptr;
                    }
                    qcp = it->second;
                }
                return qcp->create(tag, qname, auto_ack, cb);
            }
            //删除一个消费者
            void remove(const std::string& qname, const std::string& tag){
                QueueConsumer::ptr qcp;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    auto it = _qconsumers.find(qname);
                    if(it == _qconsumers.end()){
                        LOG("没有找到%s队列的消费者管理句柄!", qname.c_str());
                        return;
                    }
                    qcp = it->second;
                }
                qcp->remove(tag);
            }
            //获取一个队列的消费者
            Consumer::ptr choose(const std::string& qname){
                QueueConsumer::ptr qcp;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    auto it = _qconsumers.find(qname);
                    if(it == _qconsumers.end()){
                        LOG("没有找到%s队列的消费者管理句柄!", qname.c_str());
                        return nullptr;
                    }
                    qcp = it->second;
                }
                return qcp->choose();
            }
            void clear(){
                std::unique_lock<std::mutex> lock(_mtx);
                _qconsumers.clear();
            }
        private:
            std::mutex _mtx;
            std::unordered_map<std::string, QueueConsumer::ptr> _qconsumers;
    };
}
#endif